#pragma once

#include <functional>
#include <memory>
#include <string>

#include "AsyncWorker.hpp"
#include "client_channel.hpp"
#include "codec/codec.h"
#include "codec/dispatcher.h"
#include "mq_proto.pb.h"
#include "muduo/base/CountDownLatch.h"
#include "muduo/base/Timestamp.h"
#include "muduo/net/Callbacks.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/TcpConnection.h"

namespace MY_MQ {

class Client_Connection {
public:
    using self_ptr = std::shared_ptr<Client_Connection>;

    Client_Connection(const std::string& Server_IP, int16_t Server_Port, const AsyncWorker::self_ptr& Woker)
        : _latch(1),
          _client(Woker->loopthread.startLoop(), muduo::net::InetAddress(Server_IP, Server_Port), "Client"),
          _dispatcher(std::bind(&Client_Connection::onUnkownMessage, this, std::placeholders::_1, std::placeholders::_2,
                                std::placeholders::_3)),
          _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
                                                           std::placeholders::_1, std::placeholders::_2,
                                                           std::placeholders::_3))),
          _worker(Woker),
          _channel_manager(std::make_shared<ClientChannelManager>()) {
        _dispatcher.registerMessageCallback<MQ_Protobuf::basicCommonResponse>(
            std::bind(&Client_Connection::BasicResponse, this, std::placeholders::_1, std::placeholders::_2,
                      std::placeholders::_3));
        _dispatcher.registerMessageCallback<MQ_Protobuf::basicConsumeResponse>(
            std::bind(&Client_Connection::consumeResponse, this, std::placeholders::_1, std::placeholders::_2,
                      std::placeholders::_3));
        _client.setConnectionCallback(std::bind(&Client_Connection::onConnection, this, std::placeholders::_1));
        _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1,
                                             std::placeholders::_2, std::placeholders::_3));
        _client.setConnectionCallback(std::bind(&Client_Connection::onConnection, this, std::placeholders::_1));
        _client.connect();
        _latch.wait();
    }

    ClientChannel::self_ptr openChannel() {
        ClientChannel::self_ptr channel = _channel_manager->create(_conn, _codec);
        bool ret = channel->OpenChannel();
        if (false == ret) {
            logd("创建信道失败");
            return nullptr;
        }
        return channel;
    }

    void closeChannel(const ClientChannel::self_ptr& Channel) {
        Channel->CloseChannel();
        _channel_manager->remoeve(Channel->cid());
    }

private:
    void BasicResponse(const muduo::net::TcpConnectionPtr& conn, const basicCommonResponsePtr& message,
                       muduo::Timestamp) {
        ClientChannel::self_ptr channel = _channel_manager->get(message->cid());
        if (nullptr == channel) {
            logd("没找到信道");
        }
        channel->putBasicReponse(message);
    }

    void consumeResponse(const muduo::net::TcpConnectionPtr& conn, const basicConsumeResponsePtr& message,
                         muduo::Timestamp) {
        ClientChannel::self_ptr channel = _channel_manager->get(message->cid());
        if (nullptr == channel) {
            logd("没找到信道");
        }
        _worker->pool.enqueue([channel, message]() {
            channel->consume(message);
        });
    }

    void onUnkownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp) {
        logd("收到未知消息");
        conn->shutdown();
    }

    void onConnection(const muduo::net::TcpConnectionPtr& conn) {
        if (conn->connected()) {
            _latch.countDown();
            _conn = conn;
        } else {
            _conn.reset();
        }
    }

private:
    muduo::CountDownLatch          _latch;
    muduo::net::TcpConnectionPtr   _conn;
    muduo::net::TcpClient          _client;
    ProtobufDispatcher             _dispatcher;
    ProtobufCodecPtr               _codec;
    AsyncWorker::self_ptr          _worker;
    ClientChannelManager::self_ptr _channel_manager;
};
};  // namespace MY_MQ